package com.zilue.hfqueue.channel;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

/**
 * Created by ZD on 2017/6/2.
 */
public class MessageHandler {
    private static final String CHARSET_NAME = "UTF-8";
    private static final String START_PREFIX = "HFQUEUE";

    private static final int START_PREFIX_LENGTH = START_PREFIX.length();
    private static final int TOPIC_LENGTH = Integer.BYTES;
    private static final int DATA_LENGTH = Integer.BYTES;
    private static final int TIMESTAMP_LENGTH = Long.BYTES;
    private static final int IP_LENGTH = 15;
    private static final int ID_LENGTH = 64;
    private static final int HEAD_TOTAL_LENGTH = START_PREFIX_LENGTH + TOPIC_LENGTH + DATA_LENGTH + TIMESTAMP_LENGTH + IP_LENGTH + ID_LENGTH;


    public static ByteBuffer byteBuffer(Message message) throws IOException {
        ByteArrayOutputStream bo = new ByteArrayOutputStream();
        ObjectOutputStream oo = new ObjectOutputStream(bo);
        oo.writeObject(message.getData());
        byte[] data_bytes = bo.toByteArray();
        int data_length = data_bytes.length;
        byte[] topic_bytes = message.getTopic().getBytes(CHARSET_NAME);
        int topic_length = topic_bytes.length;
        ByteBuffer byteBuffer = ByteBuffer.allocate(HEAD_TOTAL_LENGTH + data_length + topic_length);
        byteBuffer.put(START_PREFIX.getBytes(CHARSET_NAME)).putInt(topic_length).putInt(data_length).putLong(message.getTimestamp())
                .put(message.getIp().getBytes(CHARSET_NAME)).put(message.getId().getBytes(CHARSET_NAME))
                .put(message.getTopic().getBytes(CHARSET_NAME)).put(data_bytes).flip();
        return byteBuffer;
    }


    public static Message parseMessage(SocketChannel socketChannel) throws MessageParseException {
        try {
            ByteBuffer byteBuffer = ByteBuffer.allocate(HEAD_TOTAL_LENGTH);
            socketChannel.read(byteBuffer);
            byteBuffer.flip(); //写转读
            //1.校验
            byte[] prefix_bytes = new byte[START_PREFIX_LENGTH];
            byteBuffer.get(prefix_bytes);
            if (!START_PREFIX.equals(new String(prefix_bytes, CHARSET_NAME)))
                throw new MessageParseException("message is illegal!");

            //2.获取长度
            int topic_length = byteBuffer.getInt();
            int data_length = byteBuffer.getInt();

            //3.获取时间戳
            Long timestamp = byteBuffer.getLong();
            //4.获取IP信息
            byte[] ip_bytes = new byte[IP_LENGTH];
            byteBuffer.get(ip_bytes);
            String ip = new String(ip_bytes, CHARSET_NAME);
            //5.获取ID信息
            byte[] id_bytes = new byte[ID_LENGTH];
            byteBuffer.get(id_bytes);
            String id = new String(id_bytes, CHARSET_NAME);
            //6.读取topic数据信息
            ByteBuffer topic_buffer = ByteBuffer.allocate(topic_length);
            socketChannel.read(topic_buffer);
            topic_buffer.flip();//写转读
            String topic = Charset.forName(CHARSET_NAME).decode(topic_buffer).toString();
            if (topic == null || "".equals(topic)) {
                throw new MessageParseException("topic is illegal!");
            }
            //7.读取数据信息
            ByteBuffer dataBuffer = ByteBuffer.allocate(data_length);
            socketChannel.read(dataBuffer);
            dataBuffer.flip();//写转读

            ByteArrayInputStream bo = new ByteArrayInputStream(dataBuffer.array());
            ObjectInputStream oo = new ObjectInputStream(bo);
            Object data_obj = oo.readObject();

            Message message = Message.builder().id(id).ip(ip).timestamp(timestamp).data((Serializable) data_obj).topic(topic).build();

            return message;
        } catch (IOException e) {
            throw new MessageParseException("Message parse failed ,check your message body!");
        } catch (ClassNotFoundException e) {
            throw new MessageParseException("Message data is illegal ,check your message body!");
        }
    }
}
